package org.apache.zookeeper.server.paxos2;

import org.apache.jute.InputArchive;
import org.apache.jute.OutputArchive;
import org.apache.log4j.Logger;

import java.io.*;
import java.util.BitSet;
import java.util.List;
import java.nio.ByteBuffer;

public class ExtraShard {

    private RandomAccessFile f;
    private static final Logger LOG = Logger.getLogger(Shard.class);


    private MetaStorage mapping = null;

    private BitSet freeList = null;
    private boolean freeListDirty = false;

    public ExtraShard() {
        try {
            f = new RandomAccessFile(Configuration.storagePath + File.separator + "extra", "rw");
            long length = Configuration.extraSize;
            if (f.length() != length) {
                LOG.info("file size does not match. create an empty file with size=" + length);
                f.setLength(length);
                byte[] tmp = new byte[Configuration.blockSize];
                for (int i = 0; i < (length / Configuration.blockSize); i++)
                    f.write(tmp);
                f.getFD().sync();
            }
            mapping = new MetaStorage(Metadata.path, "extramapping");
            File f = new File(Metadata.path + File.separator + "extrafree");
            if (f.exists()) {
                try {
                    ObjectInputStream ois = new ObjectInputStream(
                            new FileInputStream(f));
                    freeList = (BitSet) (ois.readObject());
                } catch (Exception e) {
                    e.printStackTrace();
                    freeList = null;
                }
            }
            if (freeList == null) {
                int totalBlocks = (int) (Configuration.extraSize / Configuration.blockSize);
                freeList = new BitSet(totalBlocks);
            }
        }
        catch (Exception e) {
            LOG.error("Couldn't create storage file", e);
            System.exit(1);
        }
    }

    public void write(long blockId, byte[] data, long version, boolean overwrite) {
        long offset;
        long blockIdInShard;

        if (overwrite) {
            byte[] physicalBlock = mapping.read(blockId);
            if(physicalBlock == null){
		mapping.print();
                throw new RuntimeException(blockId+" not found in extra");
	    }
            blockIdInShard = Metadata.bytesToLong(physicalBlock);
        } else {
            blockIdInShard = freeList.nextClearBit(0);
            if (blockIdInShard < 0 || blockIdInShard >= freeList.size()) {
                throw new RuntimeException("Extra is full");
            }
	    //System.out.println("put "+blockId+" "+blockIdInShard+" to mapping");
	    byte []haha = new byte[16];
	    ByteBuffer b = ByteBuffer.wrap(haha);
	    b.putLong(blockIdInShard);
            mapping.write(blockId, haha);//Metadata.longToBytes(blockIdInShard));
            freeList.set((int)blockIdInShard);
            freeListDirty = true;
        }
        offset = blockIdInShard * Configuration.blockSize;
        //System.out.println("extra write "+blockId+" to"+blockIdInShard);
        try {
            f.seek(offset);
            f.write(data);
        }
        catch (Exception e) {
            LOG.error("Couldn't write blockId=" + blockId, e);
            System.exit(1);
        }
    }

    public void read(long blockId, byte[] data) {
        byte[] physicalBlock = mapping.read(blockId);
        if (physicalBlock == null) {
            throw new RuntimeException("Invalid read in extra " + blockId);
        }
        long offset = Metadata.bytesToLong(physicalBlock) * Configuration.blockSize;
        try {
            f.seek(offset);
            f.readFully(data);
        }
        catch (Exception e) {
            LOG.error("Couldn't read blockId=" + blockId, e);
            System.exit(1);
        }
    }


    public void remove(long blockId) {
        byte[] physicalBlock = mapping.read(blockId);
        if (physicalBlock == null) {
            return;
        }
        freeList.clear((int) Metadata.bytesToLong(physicalBlock));
        freeListDirty = true;
    }

    public void flush() {
        try {
            f.getFD().sync();
        }
        catch (Exception e) {
            LOG.error("Couldn't flush", e);
            System.exit(1);
        }
    }


    public void serialize(OutputArchive oa, String tag) throws IOException {
        mapping.flushFile();

        if (freeListDirty) {
            try {
                ObjectOutputStream oos = new ObjectOutputStream(
                        new FileOutputStream(Metadata.path + File.separator + "extrafree"));
                oos.writeObject(freeList);
                oos.close();
            } catch (IOException e) {
                e.printStackTrace();
                System.exit(-1);
            }
            freeListDirty = false;
        }

        flush();
    }

    public void deserialize(InputArchive ia, String tag) throws IOException {

    }

    public void close() {
        try {
            f.close();
        }
        catch (Exception e) {
            LOG.error("Couldn't close", e);
        }
    }
}
